package com.xiaoniu.imclient.networkModule;


import android.text.TextUtils;

import com.xiaoniu.lib_keleim.IMClient;
import com.xiaoniu.lib_keleim.Schemes;
import com.xiaoniu.lib_keleim.pojo.vos.BaseMessage;
import com.xiaoniu.lib_keleim.pojo.vos.PreSendMessage;
import com.xiaoniu.lib_keleim.pojo.vos.UserInfoVO;
import com.xiaoniu.lib_keleim.spi.INetworkModule;
import com.xiaoniu.lib_keleim.spi.INetworkModuleFactory;
import com.xiaoniu.lib_keleim.spi.ReceiveMessageListener;
import com.xiaoniu.lib_keleim.utils.LogUtils;

import org.eclipse.paho.android.service.MqttAndroidClient;
import org.eclipse.paho.client.mqttv3.DisconnectedBufferOptions;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.util.HashSet;

public class MqttNetworkModuleFactory implements INetworkModuleFactory {

    @Override
    public Schemes getSupportedUriSchemes() {
        return Schemes.MQTT;
    }

    @Override
    public INetworkModule createNetworkModule() {
        return new MqttNetworkModule();
    }


    public static class MqttNetworkModule extends WrapperNetworkModule {
        private MqttAndroidClient mqttAndroidClient;
        private final String serverUri = "tcp://192.168.0.106:1883";
        private String clientId = "";
        HashSet<String> mTopicList = new HashSet<>();
        ReceiveMessageListener receiveMessageListener;
        @Override
        public void connect(UserInfoVO userInfo) {
            clientId = "clientId" + System.currentTimeMillis();

            mqttAndroidClient = new MqttAndroidClient(IMClient.getManager().getContext(), serverUri, clientId);
            mqttAndroidClient.setCallback(new MqttCallbackExtended() {
                @Override
                public void connectComplete(boolean reconnect, String serverURI) {

                    if (reconnect) {
                        LogUtils.d("Reconnected to : " + serverURI);
                        // Because Clean Session is true, we need to re-subscribe
                        subscribeToAllTopic();
                    } else {
                        LogUtils.d("Connected to: " + serverURI);
                    }
                }

                @Override
                public void connectionLost(Throwable cause) {
                    LogUtils.d("The Connection was lost.");
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    String receiveMessage = new String(message.getPayload());
                    LogUtils.d("Incoming message: " + new String(message.getPayload()));
                    onReceivedMessage(topic,receiveMessage);
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    LogUtils.d("deliveryComplete: ");
                }
            });

            MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
            mqttConnectOptions.setAutomaticReconnect(true);
            mqttConnectOptions.setCleanSession(false);
            mqttConnectOptions.setKeepAliveInterval(60);

            try {
                LogUtils.d("Connecting to " + serverUri);
                mqttAndroidClient.connect(mqttConnectOptions, null, new IMqttActionListener() {
                    @Override
                    public void onSuccess(IMqttToken asyncActionToken) {
                        LogUtils.d("onSuccess connect ");
                        DisconnectedBufferOptions disconnectedBufferOptions = new DisconnectedBufferOptions();
                        disconnectedBufferOptions.setBufferEnabled(true);
                        disconnectedBufferOptions.setBufferSize(100);
                        disconnectedBufferOptions.setPersistBuffer(false);
                        disconnectedBufferOptions.setDeleteOldestMessages(false);
                        mqttAndroidClient.setBufferOpts(disconnectedBufferOptions);
                        joinRoom("sys/all");
                    }

                    @Override
                    public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                        LogUtils.d("Failed to connect to: " + serverUri);
                        LogUtils.d("Failed to connect to: " + exception.getMessage() +" "+exception.getCause());
                    }
                });


            } catch (MqttException ex){
                ex.printStackTrace();
            }
        }

        public void subscribeToAllTopic(){
            for (String topic : mTopicList) {
                joinRoom(topic);
            }
        }

        @Override
        public void joinRoom(String targetId) {
            try {
                LogUtils.d("subscribeToTopic! "+targetId);
                mTopicList.add(targetId);
                mqttAndroidClient.subscribe(targetId, 0, targetId, new IMqttActionListener() {
                    @Override
                    public void onSuccess(IMqttToken asyncActionToken) {
                        LogUtils.d("Subscribed! "+targetId);
                    }
                    @Override
                    public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                        LogUtils.d("Failed to subscribe");
                    }
                });
            } catch (MqttException ex){
                LogUtils.d("Exception whilst subscribing");
                ex.printStackTrace();
            }
        }

        @Override
        public void quitRoom(String targetId) {
            LogUtils.d("unSubscribeTopic! "+targetId);
            if (mqttAndroidClient==null || TextUtils.isEmpty(targetId)) {
                return;
            }
            mTopicList.remove(targetId);
            try {
                mqttAndroidClient.unsubscribe(targetId);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void insertMessage(BaseMessage message) {

        }

        @Override
        public void sendMessage(PreSendMessage message) {
            if (mqttAndroidClient==null || message==null || TextUtils.isEmpty(message.targetId)) {
                return;
            }
            try {
                String publishTopic = "room/"+message.targetId;

                MqttMessage mqttMessage = new MqttMessage();
                mqttMessage.setPayload(message.rawMessage.getBytes());
                mqttAndroidClient.publish(publishTopic, mqttMessage);
                LogUtils.d("Message Published");
                if(!mqttAndroidClient.isConnected()){
                    mqttAndroidClient.connect();
                    LogUtils.d(mqttAndroidClient.getBufferedMessageCount() + " messages in buffer.");
                }
            } catch (MqttException e) {
                LogUtils.d("Error Publishing: " + e.getMessage());
                e.printStackTrace();
            } catch (Exception e) {
                LogUtils.d("Error Publishing: " + e.getMessage());
                e.printStackTrace();
            }
        }

        @Override
        public void addOnReceiveMessageListener(ReceiveMessageListener receiveMessageListener) {
            this.receiveMessageListener = receiveMessageListener;
        }

        public void onReceivedMessage(String target, String message) {
            receiveMessageListener.onReceivedMessage(target,message);
        }
    }

}
